package com.shujia.card

import java.sql.{Connection, PreparedStatement}

import com.alibaba.fastjson.{JSON, JSONObject}
import com.shujia.flink.FlinkTool
import com.shujia.util.{CarUtil, DateUtil}
import org.apache.commons.dbcp2.BasicDataSource
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala._

/**
  *
  * 1.1 实时统计每个卡扣当天总车流量
  *
  * 卡口,日期,车流量
  *
  */
object RealTimeCardDayFlow extends FlinkTool {

  override def run(args: Array[String]): Unit = {


    //获取过车数据
    val carsDS: DataStream[String] = CarUtil.loadKafkaData(env)

    /**
      * 解析json格式的数据
      *
      */
    val cardAndDayDS: DataStream[(Long, String, Int)] = carsDS.map(car => {

      //将json字符串转换成json对象
      val carJson: JSONObject = JSON.parseObject(car)

      //获取卡口编号
      val card: Long = carJson.getLong("card")
      //获取时间
      val time: Long = carJson.getLong("time")

      //将时间戳转换成日期
      val day: String = DateUtil.tsToDate(time)

      (card, day, 1)
    })

    /**
      * 统计每隔卡口车流量
      *
      */

    //安装卡口和日期分组
    val keyByDS: KeyedStream[(Long, String, Int), (Long, String)] = cardAndDayDS.keyBy(kv => (kv._1, kv._2))

    //统计车流量

    val sumFlowDS: DataStream[(Long, String, Int)] = keyByDS.sum(2)


    /**
      *
      * 将计算结果保存到mysql中
      *
      */
    sumFlowDS.addSink(new RichSinkFunction[(Long, String, Int)] {

      var dataSource: BasicDataSource = _

      /**
        * 在open中创建链接
        */
      override def open(parameters: Configuration): Unit = {

        //创建连接池
        dataSource = new BasicDataSource

        //设置参数
        dataSource.setUrl("jdbc:mysql://master:3306/car_ads?useUnicode=true&characterEncoding=utf-8")
        dataSource.setDriverClassName("com.mysql.jdbc.Driver")
        dataSource.setUsername("root")
        dataSource.setPassword("123456")

        //连接池初始链接数
        dataSource.setInitialSize(2)

      }

      /**
        * 回收资源
        *
        */
      override def close(): Unit = {
        dataSource.close()
      }

      /**
        * 保存数据
        *
        */
      override def invoke(value: (Long, String, Int), context: SinkFunction.Context[_]): Unit = {

        try {
          val con: Connection = dataSource.getConnection

          val stat: PreparedStatement = con.prepareStatement("replace into real_time_card_day_flow(card,day,flow) values(?,?,?)")

          stat.setLong(1, value._1)
          stat.setString(2, value._2)
          stat.setLong(3, value._3)

          stat.execute()

          con.close()
        } catch {
          case e: Exception =>

            Thread.sleep(1000)
            invoke(value, context)
        }


      }
    })

    env.execute("RealTimeCardDayFlow")
  }


}
